package com.qkjia.producer;

/**
 * @Auther 赵乾坤
 * @Date 2021-10-01 17:21
 * 带回调的生产者 异步发送
 */
import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class CallBackProducer {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1.创建kafka生产者的配置信息
        Properties properties = new Properties();

        // 添加kafka配置信息
//        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");
//        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "8.135.84.118:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
      
        // 剩下的不写都是用的默认值 
        
        // 2.创建生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);

        // 3.发送数据
        for (int i = 0; i < 10; i++) {
            // 方式1匿名表达式
        /*    producer.send(new ProducerRecord<String, String>("first",
                    "atguigu--" + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception == null) {
                        System.out.println(metadata.partition() + 
                                "--" + metadata.offset());
                    }
                }
            });*/
            // 方式2 兰布达
            // key 可以分区
            producer.send(new ProducerRecord<String, String>("first",
                    "atguigu","atguigu--" + i),(metadata,exception)->{
                if (exception == null) {
                    System.out.println(metadata.partition() +
                            "--" + metadata.offset());
                } else {
                    // 打印错误信息
                    exception.printStackTrace();
                }
            });
        }

        // 4.关闭资源
        producer.close();
    }
}